package org.liqingye.es.config;

import java.io.IOException;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ElasticsearchConfig {
    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchConfig.class);

    @Bean
    public RestClientBuilder restClientBuilder(EsProperties esProperties, HttpProperties httpProperties) {

        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(esProperties.getUsername(), esProperties.getPassword()));
        // 获取es节点的host信息
        String[] clusterNodes = esProperties.getClusterNodes().replace(" ", "").split(",");
        HttpHost[] httpHosts = new HttpHost[clusterNodes.length];
        for (int i = 0; i < clusterNodes.length; i++) {
            String[] node = clusterNodes[i].split(":");
            httpHosts[i] = new HttpHost(node[0], Integer.parseInt(node[1]));
        }
        RestClientBuilder.HttpClientConfigCallback httpClientConfigCallback = httpClientBuilder -> httpClientBuilder
                .setDefaultIOReactorConfig(IOReactorConfig.custom()
                        .setIoThreadCount(Runtime.getRuntime().availableProcessors() * 5).build())
                .setMaxConnTotal(httpProperties.getMaxTotal())// .setDefaultCredentialsProvider(credentialsProvider)
                .setMaxConnPerRoute(httpProperties.getDefaultMaxPerRoute());

        return RestClient.builder(httpHosts)
                // .setMaxRetryTimeoutMillis(httpProperties.getReadTimeout())
                .setHttpClientConfigCallback(httpClientConfigCallback)
                .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                        .setConnectionRequestTimeout(httpProperties.getConnectionRequestTimeout())
                        .setConnectTimeout(httpProperties.getConnectTimeout())
                        .setSocketTimeout(httpProperties.getReadTimeout()))
                .setFailureListener(new RestClient.FailureListener() {

                    @Override
                    public void onFailure(Node node) {
                        logger.info("elasticsearch server occur error.");
                        super.onFailure(node);
                    }
                });
    }

    @Bean
    public RestHighLevelClient restHighLevelClient(RestClientBuilder restClientBuilder) throws IOException {
        RestHighLevelClient client = new RestHighLevelClient(restClientBuilder);
        boolean ping = client.ping(RequestOptions.DEFAULT);
        if (ping) {
            logger.info("elasticesearch server connected");
        }
        return client;
    }
}
